package com.naiterui.ehp.bs.im.network.mqtt.pool;

import cn.hutool.system.HostInfo;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MqttConnectionFactory extends BasePooledObjectFactory<MqttConnection> {

    private static final Logger LOGGER = LoggerFactory.getLogger(MqttConnectionFactory.class);

    /**
     * 连接地址
     */
    private String serverURI;

    /**
     * 当前服务节点IP ：<code>localHostIP</code>
     */
    private String localHostIP;

    /**
     * mqtt连接配置
     */
    private MqttConnectOptions mqttConnectConfig;

    /**
     * 计数器
     */
    private AtomicInteger counter = new AtomicInteger();

    /**
     * 计数器锁
     */
    private Lock counterLock = new ReentrantLock();

    public MqttConnectionFactory() {
    }

    /**
     * 根据mqtt连接 配置创建工厂
     */
    public MqttConnectionFactory(String serverURI, MqttConnectOptions mqttConnectConfig) {
        this.serverURI = serverURI;
        this.mqttConnectConfig = mqttConnectConfig;
    }

    /**
     * 创建mqtt连接
     */
    @Override
    public MqttConnection create() throws Exception {

        // 实现线程安全避免在高并发的场景下出现clientId重复导致无法创建连接的情况
        int count = this.counter.addAndGet(1);

        // 根据ip+编号,生成唯一clientId
        String clientId = this.getLosthostIp() + "_" + count;

        // 生成MQTT连接
        MqttClient mqttClient = this.createMqttClient(this.serverURI, clientId, this.mqttConnectConfig);

        // 构建mqttConnection对象
        MqttConnection mqttConnection = new MqttConnection(mqttClient);
        return mqttConnection;
    }

    /**
     * mqtt连接是否可用检查
     */
    @Override
    public boolean validateObject(PooledObject<MqttConnection> pooledObject) {

        MqttConnection mqttConnection = pooledObject.getObject();
        boolean result = mqttConnection.getMqttClient().isConnected();
        LOGGER.debug("----------doValidate mqttConnectionState serverURI {},client_id {},result {}", mqttConnection.getMqttClient().getServerURI(),
                     mqttConnection.getMqttClient().getClientId(), result);
        return result;
    }

    /**
     * mqtt失效链接销毁
     */
    @Override
    public void destroyObject(PooledObject<MqttConnection> pooledObject) throws Exception {

        MqttConnection mqttConnection = pooledObject.getObject();
        LOGGER.debug("----------destroyObject mqttConnectionState ,serverURI {},client_Id {}", mqttConnection.getMqttClient().getServerURI(),
                     mqttConnection.getMqttClient().getClientId());
        mqttConnection.destroy();
    }

    /**
     * 获取当前服务真实IP
     */
    private String getLosthostIp() {
        if (StringUtils.isNotBlank(this.localHostIP)) {
            return this.localHostIP;
        }
        HostInfo hostInfo = new HostInfo();
        this.localHostIP = hostInfo.getAddress();
        return this.localHostIP;
    }

    /**
     * 封装默认返回类型
     */
    @Override
    public PooledObject<MqttConnection> wrap(MqttConnection mqttConnection) {
        return new DefaultPooledObject<>(mqttConnection);
    }

    /**
     * 创建mqtt
     */
    public MqttClient createMqttClient(String serverURI, String clientId, MqttConnectOptions config) {
        try {
            LOGGER.info("serverURI：{}，clientId：{}，config：{}，检测user.dir目录位置:{}", serverURI, clientId, config, System.getProperty("user.dir"));
            // 创建MQTT连接对象
            MqttClient mqttClient = new MqttClient(serverURI, clientId);

            // 建立连接
            mqttClient.connect(config);
            // 返回连接
            return mqttClient;

        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
            return null;
        }
    }

}
